package com.linys.scala.flink

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time

/**
  * flink实现流式处理word_count
  * @窗口函数实现
  *   依赖nc
  */
object WordCountStream {

  def main(args: Array[String]): Unit = {
    // 获取NetCat的port
    val port: Int = try{
      ParameterTool.fromArgs(args).getInt("port")
    } catch {
      case e: Exception => {
        System.err.println("No port set, use default 9000")
      }
      9000
    }

    // flink上下文环境（初始化对象）
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 获取数据
    val data = env.socketTextStream("122.51.117.240", port)

    // 隐式转换
    import org.apache.flink.api.scala._

    // 进行解析分组聚合统计
    val words = data.flatMap(line => line.split("\\s")).filter(!_.isEmpty)
    val tuples = words.map(w => WordWithCount(w, 1))
    val grouped = tuples.keyBy("word")
    val window = grouped.timeWindow(Time.seconds(2), Time.seconds(2))
    val res = window.reduce((a, b) => WordWithCount(a.word, a.count + b.count))
    res.print().setParallelism(1)

    // 执行！！！
    env.execute("first flink demo")

  }

  case class WordWithCount(word: String, count: Int)

}
